/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package redis.pipe;

/**
 *
 * @author dell
 */
import redis.clients.jedis.Jedis;
import conf.XmlUtil;
import entity.support.KeyValue;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import java.sql.ResultSet;
import java.sql.PreparedStatement;
import java.sql.DriverManager;
import java.sql.Connection;
import java.sql.ResultSetMetaData;
import java.util.List;
import java.util.LinkedList;
import java.util.Map;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
public class MPipe extends Pipe
{
//columns_______________________________________________________________________
private String driver;//the driver of mysql    
private String mysqluser;//the user of mysql
private String mysqlpassowrd;//the password of mysql  
private Connection con;//the connection to mysql
private Thread[] pipethread;//the pipethread to trans info
private CountDownLatch cdl;//manage the threads
//functions_____________________________________________________________________    
     //constructor______________________________________________________________
public MPipe()
{
     
}
     //getter___________________________________________________________________
public String getDriver()
{
     return driver;
}
public String getMysqluser()
{
     return mysqluser;
}
public String getMysqlpassword()
{
     return mysqlpassowrd;
}
     //setter___________________________________________________________________
public void setDriver(String driver)
{
     this.driver=driver;
}
public void setMysqluser(String mysqluser)
{
     this.mysqluser=mysqluser;
}
public void setMysqlpassword(String mysqlpassword)
{
     this.mysqlpassowrd=mysqlpassword;
}
     //operator_________________________________________________________________
         //init functions_______________________________________________________
/**
 * asign value for propertites related to Redis
 * @param key
 * @param value
 * @throws Exception 
 * @pass
 */
private void setRedis(String key,String value) throws Exception
{
     switch(key)
     {
         case "redisHost":this.redishost=value;break;
         case "redisPort":this.redisport=Integer.valueOf(value);break;
         case "redisPassword":this.redispassword=value;break;
         default:throw new Exception("Unknown tag property in redis:"+key);
     }
}
/**
 * asign value for propertites related to MySQL
 * @param key
 * @param value
 * @throws Exception 
 * @pass
 */
private void setMysql(String key,String value) throws Exception
{
     switch(key)
     {
         case "driver":this.driver=value;break;
         case "dbUrl":this.dburl=value;break;
         case "mysqlUser":this.mysqluser=value;break;
         case "mysqlPassword":this.mysqlpassowrd=value;break;
         default:throw new Exception("Unknown tag property in mysql:"+key);
     }
     
}
/**
 * asign values for maps from tables in mysql to databases in redis,
 * tablename is stored in key,while target is stored in value
 * @param list
 * @throws Exception 
 * @pass
 */
private void setPipes(NodeList list) throws Exception
{
     this.pipes=new LinkedList<KeyValue<String,Integer>>();
     List<String> tables=new LinkedList<String>();
     List<Integer> targets=new LinkedList<Integer>();
     for(int i=0;i<list.getLength();i++)//pipe
     {
         Node n=list.item(i);
         if(n.getNodeType()==Node.ELEMENT_NODE)
         {
             NodeList m=n.getChildNodes();
             for(int j=0;j<m.getLength();j++)
             {
                 Node k=m.item(j);
                 if(k.getNodeType()==Node.ELEMENT_NODE)
                 {
                     String name=k.getNodeName();
                     String value=k.getFirstChild().getNodeValue();
                     if("table".equals(name)) tables.add(value);
                     else if("target".equals(name)) targets.add(Integer.valueOf(value));
                 }
             }
         }
     }
     Iterator<String> p1=tables.iterator();
     Iterator<Integer> p2=targets.iterator();
     while(p1.hasNext()&&p2.hasNext())
     {
          String table=p1.next();
          Integer target=p2.next();
          this.pipes.add(new KeyValue(table,target));
     }
}
/**
 * asign values to properties of this class
 * @param url
 * @throws Exception 
 * @pass
 */
private void initProperty(String url) throws Exception
{
     Document document=XmlUtil.getDocument(url);
     NodeList list=document.getElementsByTagName("configuration").item(0).getChildNodes();
     NodeList mysql=null,redis=null,pipe=null;
     for(int i=0;i<list.getLength();i++)
     {
         Node n=list.item(i);
         if(n.getNodeType()==Element.ELEMENT_NODE)
         {
             String name=n.getNodeName();
             switch(name)
             {
                 case "mysql":mysql=n.getChildNodes();break;
                 case "redis":redis=n.getChildNodes();break;
                 case "pipes":pipe=n.getChildNodes();break;
                 default:throw new Exception("Unknown tag property:"+name);
             }
         }
     }
     for(int j=0;j<mysql.getLength();j++)
     {
         Node k=mysql.item(j);
         if(k.getNodeType()==Node.ELEMENT_NODE)
         {
             String key=k.getNodeName();
             String value=k.getFirstChild().getNodeValue();
             this.setMysql(key, value);
         }   
     }
     for(int j=0;j<redis.getLength();j++)
     {
         Node k=redis.item(j);
         if(k.getNodeType()==Node.ELEMENT_NODE)
         {
             String key=k.getNodeName();
             String value=k.getFirstChild().getNodeValue();
             this.setRedis(key, value);
         }
     }
     this.setPipes(pipe);
}
/**
 * init environment of mysql
 * @throws Exception 
 * @pass
 */
private void initMySQL() throws Exception
{ 
     Class.forName(this.driver);
     con=DriverManager.getConnection(this.dburl, this.mysqluser, this.mysqlpassowrd);
}
/**
 * run method initProperty(),initMySQL,init Redis() in sequence
 * @param url
 * @return 
 * @pass
 */
@Override
protected void init(String url) 
{
     try
     {
         this.initProperty(url);
         this.initMySQL();
         this.initialized=true;
     }
     catch(Exception e)
     {
         e.printStackTrace();
     }
}
         //pipe functions_______________________________________________________
/**
 * get the detail information of the specified tabele,and perse 
 * infomation in the Table Object
 * @param pst
 * @param tablename
 * @throws Exception
 * @pass
 */
private Table getTable(String tablename,int target) 
{
     Table table=null;
     String query="select * from "+tablename;
     PreparedStatement pst=null;
     try
     {
         pst=con.prepareStatement(query);
         ResultSetMetaData rsmd=pst.getMetaData();
         int len=rsmd.getColumnCount();
         table=new Table(tablename,target,len);
         for(int i=0;i<len;i++)
         {
             String name=rsmd.getColumnLabel(i+1);
             String javatype=rsmd.getColumnClassName(i+1);
             String jdbctype=rsmd.getColumnTypeName(i+1);
             if(name!=null&&javatype!=null&&jdbctype!=null)
             {
                 table.columns[i]=new Table.Column(name, jdbctype, javatype);
             }
         }
     }
     catch(Exception e)
     {
          e.printStackTrace();
     }
     finally
     {
         try
         {
             pst.close();
         }
         catch(Exception e)
         {
             e.printStackTrace();
         }
     }
     return table;
}
/**
 * return a jedis connect to Redis
 * @throws Exception 
 * @pass
 */
private Jedis getRedis()
{
     Jedis jedis=null;
     try
     {
         jedis=new Jedis(this.redishost,this.redisport);
         jedis.auth(this.redispassword);
         if(!jedis.isConnected()) throw new Exception("error in initialization of Redis");
     }
     catch(Exception e)
     { 
         e.printStackTrace();
     }
     return jedis;
}
/**
 * give the Table ,PreparedStatement and Jedis,to get a Thread instance
 * @param table
 * @param pst
 * @return
 * @throws Exception 
 * @pass
 */
private Thread getPipeThread(KeyValue<String,Integer> kv,Jedis jedis) throws Exception
{
     Thread t=null;
     String tablenames[]=kv.getKey().split(",");
     int target=kv.getValue();
     Table tables[]=new Table[tablenames.length];
     for(int i=0;i<tablenames.length;i++) tables[i]=this.getTable(tablenames[i], target);
     t=new Thread(new Runnable()
     {
         private PreparedStatement pst;
         private ResultSet rs;
         @Override
         public void run() 
         {
             try
             {
                 String query="select * from "+tables[0].tablename;
                 for(int i=1;i<tables.length;i++) query+=","+tables[i].tablename;
                 jedis.select(tables[0].getTarget());
                 pst=MPipe.this.con.prepareCall(query); 
                 rs=pst.executeQuery();
                 while(rs.next())
                 {
                     Map<String,String> map=new HashMap<String,String>();
                     String key=rs.getString(tables[0].columns[0].getColumnname())+"*"+
                                rs.getString(tables[0].columns[1].getColumnname());
                     for(int i=0;i<tables[0].columns.length;i++)
                     {
                         String field=tables[0].columns[i].columnname;
                         String value=rs.getString(field);
                         if(field!=null&&value!=null) map.put(field, value);
                      }
                     if(key!=null&&!map.isEmpty()) jedis.hmset(key, map);
                 }
             }
             catch(Exception e)
             {
                 e.printStackTrace();
             }
             finally
             {
                 try
                 {
                     pst.close();
                     rs.close();
                     MPipe.this.cdl.countDown();
                  }
                 catch(Exception e)
                 {
                     e.printStackTrace();
                  }
             }
         }
     });
     return t;
}
/**
 * init and start all threads,to trans info from mysql to redis
 * @return 
 * @pass
 */
@Override
protected boolean pipe() 
{
     boolean result=false;
     try
     {
         pipethread=new Thread[pipes.size()];
         cdl=new CountDownLatch(pipes.size());
         Iterator<KeyValue<String,Integer>> iter=pipes.iterator();
         for(int i=0;iter.hasNext()&&i<pipethread.length;i++)
         {
             KeyValue<String,Integer> kv=iter.next();
             pipethread[i]=this.getPipeThread(kv, this.getRedis());
         }
         for(int i=0;i<pipethread.length;i++)
         {
             if(pipethread[i]!=null) pipethread[i].start();
         }
         result=true;
     }
     catch(Exception e)
     {
         e.printStackTrace();
     }
     return result;
}
         //destroy functions____________________________________________________
@Override
protected boolean destroy() 
{
     boolean result=false;
     try
     {
         cdl.await();
         this.con.close();
         result=true;
     }
     catch(Exception e)
     {
         e.printStackTrace();
     }
     return result;
}   
/**
 * run methods init,pipe,destroy in sequence
 * @param url 
 * @pass
 */
public void work(String url)
{
     this.init(url);
     if(!this.initialized) System.err.println("init failed");
     if(!this.pipe()) System.err.println("pipe failed");
     if(!this.destroy()) System.err.println("destroy failed");
}
     //toString_________________________________________________________________
@Override
public String toString()
{
     return super.toString()+"\t"+driver+"\t"+mysqluser+"\t"+mysqlpassowrd+"\t";
}
}
